package spring.cloud.tasks.tasks_assign_manager.service;

import com.alibaba.fastjson.JSON;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import spring.cloud.tasks.common.node.BlockNodePaths;
import spring.cloud.tasks.common.task_item_allot.OnlineExecutor;
import spring.cloud.tasks.common.task_item_allot.TaskItem;
import spring.cloud.tasks.tasks_assign_manager.database.entity.TaskItemAssignEventDto;
import spring.cloud.tasks.tasks_assign_manager.task_item_assign.NameSpaceTasksTaskItemAssign;
import spring.cloud.tasks.tasks_assign_manager.task_item_assign.pick_up.executor.PickUpWhenExecutorLostTraffic;
import spring.cloud.tasks.tasks_assign_manager.task_item_assign.pick_up.executor.PickUpWhenExecutorOffline;
import spring.cloud.tasks.tasks_assign_manager.task_item_assign.pick_up.executor.PickUpWhenExecutorOnline;
import spring.cloud.tasks.tasks_assign_manager.task_item_assign.pick_up.executor.PickUpWhenExecutorRecoverTraffic;
import spring.cloud.tasks.tasks_assign_manager.task_item_assign.pick_up.executor.task.PickUpWhenTaskExecutorOffline;
import spring.cloud.tasks.tasks_assign_manager.task_item_assign.pick_up.executor.task.PickUpWhenTaskExecutorOnline;
import spring.cloud.tasks.tasks_assign_manager.task_item_assign.pick_up.task.PickUpWhenTaskDisable;
import spring.cloud.tasks.tasks_assign_manager.task_item_assign.pick_up.task.PickUpWhenTaskEnable;
import spring.cloud.tasks.tasks_assign_manager.task_item_assign.pick_up.task.TaskOneKeyTaskItemAssign;
import spring.cloud.tasks.tasks_assign_manager.task_item_assign_event.*;
import spring.cloud.tasks.tasks_assign_manager.task_item_assign_event_handler.*;

import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.concurrent.atomic.AtomicBoolean;

@Slf4j
@Setter
@Getter
public class TaskItemAssignService {
    private String nameSpace;
    private CuratorFramework curatorFramework;
    private AtomicBoolean needAllSharding;
    private String host;

    public TaskItemAssignService(CuratorFramework curatorFramework, String host) {
        this.curatorFramework = curatorFramework;
        this.nameSpace = curatorFramework.getNamespace();
        this.needAllSharding = new AtomicBoolean(false);
        this.host = host;
    }

    //##<!-我是优美的功能模块分割线->##
    public void fixTaskItemIndexNode(List<OnlineExecutor> onlineExecutorList) throws Exception {
        String taskItemNodesTasksNodePath = BlockNodePaths.getTaskItemNodesTasksNodePath();
        if (curatorFramework.checkExists().forPath(taskItemNodesTasksNodePath) == null) {
            curatorFramework.create().creatingParentsIfNeeded().forPath(taskItemNodesTasksNodePath);
        }
        List<String> oldTaskItemIndexPathList = getTaskItemIndexPathListList(taskItemNodesTasksNodePath);
        //
        List<String> newTaskItemIndexPathList = new ArrayList<String>();
        for (OnlineExecutor onlineExecutor : onlineExecutorList) {
            for (TaskItem taskItem : onlineExecutor.getAssignedTaskItemList()) {
                String taskId = taskItem.getTaskId();
                String taskInstanceId = taskItem.getTaskInstanceId();
                String taskRoundId = taskItem.getTaskRoundId();
                int taskItemIndex = taskItem.getTaskItemIndex();
                String taskItemIndexPath = BlockNodePaths.getTaskItemNodesTaskItemNodePath(taskId, taskInstanceId, taskRoundId, taskItemIndex + "");
                newTaskItemIndexPathList.add(taskItemIndexPath);
            }
        }
        //找到需要新增的
        //找到需要删除的
        List<String> addList = new ArrayList<String>();
        List<String> removeList = new ArrayList<String>();
        for (String newTaskItemIndexPath : newTaskItemIndexPathList) {
            if (!oldTaskItemIndexPathList.contains(newTaskItemIndexPath)) {
                addList.add(newTaskItemIndexPath);
            }
        }
        for (String oldTaskItemIndexPath : oldTaskItemIndexPathList) {
            if (!newTaskItemIndexPathList.contains(oldTaskItemIndexPath)) {
                removeList.add(oldTaskItemIndexPath);
            }
        }
        Map<String, TaskItem> taskItemMap = new HashMap<String, TaskItem>();
        for (String add : addList) {
            for (OnlineExecutor onlineExecutor : onlineExecutorList) {
                for (TaskItem taskItem : onlineExecutor.getAssignedTaskItemList()) {
                    String taskId = taskItem.getTaskId();
                    String taskInstanceId = taskItem.getTaskInstanceId();
                    String taskRoundId = taskItem.getTaskRoundId();
                    int taskItemIndex = taskItem.getTaskItemIndex();
                    String taskItemIndexPath = BlockNodePaths.getTaskItemNodesTaskItemNodePath(taskId, taskInstanceId, taskRoundId, taskItemIndex + "");
                    if (add.equals(taskItemIndexPath)) {
                        taskItemMap.put(taskItemIndexPath, taskItem);
                    }
                }
            }
        }
        for (String remove : removeList) {
            curatorFramework.delete().forPath(remove);
        }
        for (String add : addList) {
            TaskItem taskItem = taskItemMap.get(add);
            curatorFramework.create().forPath(add, JSON.toJSONString(taskItem).getBytes(StandardCharsets.UTF_8));
        }
    }

    private List<String> getTaskItemIndexPathListList(String taskItemNodesTasksNodePath) throws Exception {
        List<String> oldTaskItemIndexPathList = new ArrayList<String>();
        List<String> oldTaskIdList = curatorFramework.getChildren().forPath(taskItemNodesTasksNodePath);
        if (oldTaskIdList != null && !oldTaskIdList.isEmpty()) {
            for (String taskId : oldTaskIdList) {
                List<String> taskInstanceIdList = curatorFramework.getChildren().forPath(BlockNodePaths.getTaskItemNodesTaskInstancesNodePath(taskId));
                if (taskInstanceIdList != null) {
                    for (String taskInstanceId : taskInstanceIdList) {
                        List<String> taskRoundIdList = curatorFramework.getChildren().forPath(BlockNodePaths.getTaskItemNodesTaskRoundIdsNodePath(taskId, taskInstanceId));
                        if (taskRoundIdList != null) {
                            for (String taskRoundId : taskRoundIdList) {
                                List<String> taskItemIndexList = curatorFramework.getChildren().forPath(BlockNodePaths.getTaskItemNodesTaskItemsNodePath(taskId, taskInstanceId, taskRoundId));
                                if (taskItemIndexList != null) {
                                    for (String taskItemIndex : taskItemIndexList) {
                                        String taskItemIndexPath = BlockNodePaths.getTaskItemNodesTaskItemNodePath(taskId, taskInstanceId, taskRoundId, taskItemIndex);
                                        oldTaskItemIndexPathList.add(taskItemIndexPath);
                                    }
                                }
                            }
                        }
                    }
                }
            }
        }
        return oldTaskItemIndexPathList;
    }

    /**
     * 持久化任务分片信息(per OnlineExecutor)
     */
    public void persistOnlineExecutorTaskItemAssignList(List<OnlineExecutor> onlineExecutorList) throws Exception {
        //如果不存在，则新建数据节点
        String nodePath = BlockNodePaths.ONLINE_EXECUTOR_TASK_ITEM_ASSIGNS_NODE_PATH;
        if (curatorFramework.checkExists().forPath(nodePath) == null) {
            curatorFramework.create().creatingParentsIfNeeded().forPath(nodePath);
        }
        //删除OLD数据节点
        List<String> childrenPathList = curatorFramework.getChildren().forPath(nodePath);
        if (childrenPathList != null && !childrenPathList.isEmpty()) {
            for (String childrenPath : childrenPathList) {
                curatorFramework.delete().forPath(BlockNodePaths.getOnlineExecutorTaskItemAssignsNodePath(childrenPath));
            }
        }
        // 持久化新数据到节点
        /**
         * 如果内容过大，使用事务提交，会报长度过长的错误。故采用非事务方式提交
         */
        for (int i = 0; i < onlineExecutorList.size(); i++) {
            byte[] bytes = JSON.toJSONString(onlineExecutorList.get(i)).getBytes(StandardCharsets.UTF_8);
            curatorFramework.create().forPath(BlockNodePaths.getOnlineExecutorTaskItemAssignsNodePath(onlineExecutorList.get(i).getExecutorId()), bytes);
        }
    }

    /**
     * 获取任务分片信息(per OnlineExecutor)
     */
    public List<OnlineExecutor> getOldOnlineExecutorTaskItemAssignList() throws Exception {
        List<OnlineExecutor> onlineExecutorList = new ArrayList<OnlineExecutor>();
        if (curatorFramework.checkExists().forPath(BlockNodePaths.ONLINE_EXECUTOR_TASK_ITEM_ASSIGNS_NODE_PATH) != null) {
            List<String> childrenPathList = curatorFramework.getChildren().forPath(BlockNodePaths.ONLINE_EXECUTOR_TASK_ITEM_ASSIGNS_NODE_PATH);
            Collections.sort(childrenPathList, new Comparator<String>() {
                @Override
                public int compare(String arg0, String arg1) {
                    Integer a = Integer.valueOf(arg0);
                    Integer b = Integer.valueOf(arg1);
                    return a.compareTo(b);
                }
            });
            for (String childrenPath : childrenPathList) {
                byte[] bytes = curatorFramework.getData().forPath(BlockNodePaths.getOnlineExecutorTaskItemAssignsNodePath(childrenPath));
                onlineExecutorList.add((OnlineExecutor) JSON.parse(bytes));
            }
        }
        return onlineExecutorList;
    }

    //##<!-我是优美的功能模块分割线->##
    public List<TaskItemAssignEvent> toTaskAllotEvent(List<TaskItemAssignEventDto> taskItemAssignEventDtoList) throws Exception {
        List<TaskItemAssignEvent> taskAllotEventList = new ArrayList<TaskItemAssignEvent>();
        for (TaskItemAssignEventDto dto : taskItemAssignEventDtoList) {
            if (NameSpaceTasksTaskItemAssignEvent.type_tasks_full_allot.equals(dto.getType())) {
                taskAllotEventList.add(new NameSpaceTasksTaskItemAssignEvent());
            } else if (TaskEnableOrDisEnableTaskItemAssignEvent.type_task_enable.equals(dto.getType()) ||//
                    TaskEnableOrDisEnableTaskItemAssignEvent.type_task_disable.equals(dto.getType()) ||//
                    TaskEnableOrDisEnableTaskItemAssignEvent.type_task_force_reallot.equals(dto.getType())//
            ) {
                taskAllotEventList.add(new TaskEnableOrDisEnableTaskItemAssignEvent(dto.getType(), dto.getTaskId()));
            } else if (ExecutorOnlineOrOfflineTaskItemAssignEvent.type_executor_online.equals(dto.getType()) ||//
                    ExecutorOnlineOrOfflineTaskItemAssignEvent.type_executor_offline.equals(dto.getType())) {//
                taskAllotEventList.add(new ExecutorOnlineOrOfflineTaskItemAssignEvent(dto.getType(), dto.getExecutorId()));
            } else if (ExecutorLostOrRecoverTrafficTaskItemAssignEvent.type_executor_lost_traffic.equals(dto.getType()) ||//
                    ExecutorLostOrRecoverTrafficTaskItemAssignEvent.type_executor_recover_traffic.equals(dto.getType())) {
                taskAllotEventList.add(new ExecutorLostOrRecoverTrafficTaskItemAssignEvent(dto.getType(), dto.getExecutorId()));
            } else if (TaskExecutorOnlineOrOfflineTaskItemAssignEvent.type_task_executor_online.equals(dto.getType()) ||
                    TaskExecutorOnlineOrOfflineTaskItemAssignEvent.type_task_executor_online.equals(dto.getType())
            ) {
                taskAllotEventList.add(new TaskExecutorOnlineOrOfflineTaskItemAssignEvent(dto.getType(), dto.getTaskId(), dto.getExecutorId()));
            } else {
                //...
            }
        }
        return taskAllotEventList;
    }

    public void handles(List<TaskItemAssignEvent> taskItemAssignEventList) throws Exception {
        for (TaskItemAssignEvent event : taskItemAssignEventList) {
            TaskItemAssignEventHandler handler = getTaskAllotHandler(event);
            if (handler != null) {
                handler.handle(event);
            }
        }
    }

    public TaskItemAssignEventHandler getTaskAllotHandler(TaskItemAssignEvent taskItemAssignEvent) {
        if (taskItemAssignEvent instanceof NameSpaceTasksTaskItemAssignEvent) {
            return new NameSpaceTasksTaskItemAssignEventHandler(this);
            //
        }
        if (taskItemAssignEvent instanceof TaskEnableOrDisEnableTaskItemAssignEvent) {
            return new TaskEnableOrDisEnableTaskItemAssignEventHandler(this);
            //
        } else if (taskItemAssignEvent instanceof ExecutorOnlineOrOfflineTaskItemAssignEvent) {
            return new ExecutorOnlineOfflineTaskItemAssignEventHandler(this);
        } else if (taskItemAssignEvent instanceof ExecutorLostOrRecoverTrafficTaskItemAssignEvent) {
            return new ExecutorLostOrRecoverTrafficTaskItemAssignEventHandler(this);
            //
        } else if (taskItemAssignEvent instanceof TaskExecutorOnlineOrOfflineTaskItemAssignEvent) {
            return new TaskExecutorOnlineOrOfflineTaskItemAssignEventHandler(this);
        } else {
            return null;
        }
    }

    /**
     * 进行全量分片
     */
    public void tasksFullAllot() throws Exception {
        {
            needAllSharding.set(true);
            new NameSpaceTasksTaskItemAssign(this);//.taskItemAssign();
            String shardAllAtOnce = BlockNodePaths.getOnlineExecutorTaskItemAssignsNodePath("shardAllAtOnce");
            if (curatorFramework.checkExists().forPath(shardAllAtOnce) != null) {
                curatorFramework.delete().deletingChildrenIfNeeded().forPath(shardAllAtOnce);
            }
        }
    }

    /**
     * 任务启用事件
     */
    public void taskEnableAllotTask(String taskId) throws Exception {
        {
            PickUpWhenTaskEnable taskEnableTaskItemAssign = new PickUpWhenTaskEnable(this);//.taskItemAssign();
            taskEnableTaskItemAssign.setTaskId(taskId);
        }
    }

    /**
     * 处理任务禁用事件
     */
    public void taskDisableAllotTask(String taskId) throws Exception {
        {

            PickUpWhenTaskDisable pickUpWhenTaskDisable = new PickUpWhenTaskDisable(this);//.taskItemAssign();
            pickUpWhenTaskDisable.setTaskId(taskId);
        }
    }

    /**
     * 处理任务全排
     */
    public void taskExecuteForceTaskItemAllotTask(String taskId) throws Exception {
        {
            TaskOneKeyTaskItemAssign taskOneKeyTaskItemAssign = new TaskOneKeyTaskItemAssign(this);//.taskItemAssign();
            taskOneKeyTaskItemAssign.setTaskId(taskId);
        }
    }

    /**
     * 结点上线处理
     */
    public void executorOnlineAllotTask(String executorId) throws Exception {
        {

            String ip = null;
            PickUpWhenExecutorOnline executorOnlineTaskItemAssign = new PickUpWhenExecutorOnline(this);//.taskItemAssign();
            executorOnlineTaskItemAssign.setExecutorId(executorId);
            executorOnlineTaskItemAssign.setIp(ip);
        }
    }

    /**
     * 结点掉线处理
     */
    public void executorOfflineAllotTask(String executorId) throws Exception {
        {

            PickUpWhenExecutorOffline executorOfflineTaskItemAssign = new PickUpWhenExecutorOffline(this);//.taskItemAssign();
            executorOfflineTaskItemAssign.setExecutorId(executorId);
        }
    }

    /**
     * 摘取流量
     */
    public void executorLostTrafficAllotTask(String executorId) throws Exception {
        {

            PickUpWhenExecutorLostTraffic executorLostTrafficTaskItemAssign = new PickUpWhenExecutorLostTraffic(this);//.taskItemAssign();
            executorLostTrafficTaskItemAssign.setExecutorId(executorId);
        }
    }

    /**
     * 恢复流量
     */
    public void executorRecoverTrafficAllotTask(String executorId) throws Exception {
        {

            PickUpWhenExecutorRecoverTraffic executorRecoverTrafficTaskItemAssign = new PickUpWhenExecutorRecoverTraffic(this);//.taskItemAssign();
            executorRecoverTrafficTaskItemAssign.setExecutorId(executorId);
        }
    }


    /**
     * 处理任务executor上线
     */
    public void taskExecutorOnlineAllotTask(String taskId, String executorId) throws Exception {
        {
            PickUpWhenTaskExecutorOnline taskExecutorOnlineTaskItemAssign = new PickUpWhenTaskExecutorOnline(this);//.taskItemAssign();
            taskExecutorOnlineTaskItemAssign.setExecutorId(executorId);
            taskExecutorOnlineTaskItemAssign.setTaskId(taskId);
        }
    }

    /**
     * 处理任务executor下线
     */
    public void taskExecutorOfflineAllotTask(String taskId, String executorId) throws Exception {
        {
            PickUpWhenTaskExecutorOffline taskExecutorOfflineTaskItemAssign = new PickUpWhenTaskExecutorOffline(this);//.taskItemAssign();
            taskExecutorOfflineTaskItemAssign.setExecutorId(executorId);
            taskExecutorOfflineTaskItemAssign.setTaskId(taskId);
        }
    }


    private void tasksFullTaskItemTask() {
        // 提交全量分片线程
        needAllSharding.set(true);
        new NameSpaceTasksTaskItemAssign(this);//.taskItemAssign();
        log.info("{}-{} become leader", nameSpace, host);
    }

    public void shutdown() throws InterruptedException {

    }

    public boolean isNeedAllSharding() {
        return needAllSharding.get();
    }

    public void setNeedAllSharding(boolean needAllSharding) {
        this.needAllSharding.set(needAllSharding);
    }


}
